/* Copyright (c) 2023 Renmin University of China
RMDB is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:
        http://license.coscl.org.cn/MulanPSL2
THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details. */

#pragma once
#include "execution_defs.h"
#include "execution_manager.h"
#include "executor_abstract.h"
#include "index/ix.h"
#include "system/sm.h"

class NestedLoopJoinExecutor : public AbstractExecutor {
private:
    std::unique_ptr<AbstractExecutor> left_;    // 左儿子节点（需要join的表）
    std::unique_ptr<AbstractExecutor> right_;   // 右儿子节点（需要join的表）
    size_t len_;                                // join后获得的每条记录的长度
    std::vector<ColMeta> cols_;                 // join后获得的记录的字段

    std::vector<Condition> fed_conds_;          // join条件
    bool isend;

    std::vector<std::unique_ptr<RmRecord>> left_buffer;
    std::vector<std::unique_ptr<RmRecord>> right_buffer;

    //缓冲区暂定为300 避免过大影响内存使用
    // 开大一点蒙一蒙八
    size_t LEFT_BUFFER_SIZE = 5000;
    size_t RIGHT_BUFFER_SIZE = 5000;
    size_t left_buffer_index;
    size_t right_buffer_index;

    //试着优化
    std::vector<std::pair<int,int>> cond_offset;
    std::vector<ColType> cond_lhs_type;

public:
    NestedLoopJoinExecutor(std::unique_ptr<AbstractExecutor> left, std::unique_ptr<AbstractExecutor> right,
        std::vector<Condition> conds) {
        left_ = std::move(left);
        right_ = std::move(right);
        len_ = left_->tupleLen() + right_->tupleLen();
        cols_ = left_->cols();
        auto right_cols = right_->cols();
        for (auto& col : right_cols) {
            col.offset += left_->tupleLen();
        }

        cols_.insert(cols_.end(), right_cols.begin(), right_cols.end());
        isend = false;
        fed_conds_ = std::move(conds);
        //预留缓存空间，初始化索引
        left_buffer.reserve(LEFT_BUFFER_SIZE);
        right_buffer.reserve(RIGHT_BUFFER_SIZE);
        left_buffer_index = 0;
        right_buffer_index = 0;

        //优化用
        for(auto & cond:fed_conds_)
        {   
            //计算左右列偏移
            auto lcolmeta = *get_col(left_->cols(), cond.lhs_col);
            auto rcolmeta = *get_col(right_->cols(), cond.rhs_col);
            cond_offset.emplace_back(lcolmeta.offset,rcolmeta.offset);
            //记录左值类型
            cond_lhs_type.emplace_back(lcolmeta.type);
        }

    }
    // //主要思路 用两个buffer表示两个表 随后buffer大小就对应一个块，随后按照嵌套块循环连接的方法实现
    // void beginTuple() override {
    //     //初始化两个缓存
    //     init_buffer();
    //     if (isend)
    //         return;
    //     while (true)
    //     {
    //         //外部块元组遍历
    //         for (; left_buffer_index < left_buffer.size(); left_buffer_index++)
    //         {
    //             //内部块元组遍历
    //             for (; right_buffer_index < right_buffer.size(); right_buffer_index++)
    //             {
    //                 if (test_join_conds(*left_buffer[left_buffer_index], *right_buffer[right_buffer_index]))
    //                 {
    //                     //找到了符合条件的第一个
    //                     return;
    //                 }
    //             }
    //             //当前内部块缓存被全部扫描，置为0
    //             right_buffer_index = 0;
    //         }

    //         left_buffer_index = 0;

    //         //内部块缓存满了，读入下一个块
    //         right_buffer.clear();

    //         while (!right_->is_end() && right_buffer.size() < RIGHT_BUFFER_SIZE)
    //         {
    //             right_buffer.emplace_back(right_->Next());
    //             right_->nextTuple();
    //         }

    //         //内部块遍历完毕，读入外部快的下一块
    //         if (right_buffer.size() == 0)
    //         {
    //             left_buffer.clear();
    //             while (!left_->is_end() && left_buffer.size() < LEFT_BUFFER_SIZE)
    //             {
    //                 left_buffer.emplace_back(left_->Next());
    //                 left_->nextTuple();
    //             }
    //             if (left_buffer.size() == 0 && left_->is_end())
    //             {
    //                 isend = true;
    //                 return;
    //             }
    //             left_buffer_index = 0;

    //             //重新读入内部块
    //             right_->beginTuple();
    //             while (!right_->is_end() && right_buffer.size() < RIGHT_BUFFER_SIZE)
    //             {
    //                 right_buffer.emplace_back(right_->Next());
    //                 right_->nextTuple();
    //             }

    //         }
    //     }

    // }

    // void nextTuple() override
    // {
    //     right_buffer_index++;
    //     while (true)
    //     {
    //         //外部块元组遍历
    //         for (; left_buffer_index < left_buffer.size(); left_buffer_index++)
    //         {
    //             //内部块元组遍历
    //             for (; right_buffer_index < right_buffer.size(); right_buffer_index++)
    //             {
    //                 if (test_join_conds(*left_buffer[left_buffer_index], *right_buffer[right_buffer_index]))
    //                 {
    //                     //找到了符合条件的第一个
    //                     return;
    //                 }
    //             }
    //             //当前内部块缓存被全部扫描，置为0
    //             right_buffer_index = 0;
    //         }

    //         left_buffer_index = 0;

    //         //内部块缓存满了，读入下一个块
    //         right_buffer.clear();

    //         while (!right_->is_end() && right_buffer.size() < RIGHT_BUFFER_SIZE)
    //         {
    //             right_buffer.emplace_back(right_->Next());
    //             right_->nextTuple();
    //         }

    //         //内部块遍历完毕，读入外部快的下一块
    //         if (right_buffer.size() == 0)
    //         {
    //             left_buffer.clear();
    //             while (!left_->is_end() && left_buffer.size() < LEFT_BUFFER_SIZE)
    //             {
    //                 left_buffer.emplace_back(left_->Next());
    //                 left_->nextTuple();
    //             }
    //             if (left_buffer.size() == 0 && left_->is_end())
    //             {
    //                 isend = true;
    //                 return;
    //             }
    //             left_buffer_index = 0;

    //             //重新读入内部块
    //             right_->beginTuple();
    //             while (!right_->is_end() && right_buffer.size() < RIGHT_BUFFER_SIZE)
    //             {
    //                 right_buffer.emplace_back(right_->Next());
    //                 right_->nextTuple();
    //             }

    //         }
    //     }
    // }
    // //用于检验连接条件
    // bool test_join_conds(RmRecord& lrecord, RmRecord& rrecord)
    // {
    //     for (auto& cond : fed_conds_)
    //     {
    //         auto lcolmeta = *get_col(left_->cols(), cond.lhs_col);
    //         char* lval = lrecord.data + lcolmeta.offset;

    //         auto rcolmeta = *get_col(right_->cols(), cond.rhs_col);
    //         char* rval = rrecord.data + rcolmeta.offset;

    //         int cmp;
    //         //比较方法与抽象类中定义的一致
    //         switch (lcolmeta.type)
    //         {
    //         case TYPE_INT:
    //         {
    //             int il = *(int*)lval;
    //             int ir = *(int*)rval;
    //             // cmp=il-ir;
    //             cmp = (il < ir) ? -1 : ((il > ir) ? 1 : 0);
    //             break;
    //         }
    //         case TYPE_BIGINT:
    //         {
    //             long long bl = *(long long*)lval;
    //             long long br = *(long long*)rval;
    //             cmp = (bl < br) ? -1 : ((bl > br) ? 1 : 0);
    //             break;
    //         }
    //         case TYPE_FLOAT:
    //         {
    //             float fl = *(float*)lval;
    //             float fr = *(float*)rval;
    //             cmp = (fl < fr) ? -1 : ((fl > fr) ? 1 : 0);
    //             break;
    //         }
    //         case TYPE_STRING:
    //         {
    //             std::string sl = *(std::string*)lval;
    //             std::string sr = *(std::string*)rval;
    //             cmp = sl.compare(sr);
    //             break;
    //         }
    //         case TYPE_TIME:
    //         {
    //             cmp = memcmp(lval, rval, lcolmeta.len);
    //         }
    //         }

    //         bool res = true;
    //         switch (cond.op)
    //         {
    //         case OP_EQ:
    //             res = cmp == 0;
    //             break;
    //         case OP_NE:
    //             res = cmp != 0;
    //             break;
    //         case OP_LT:
    //             res = cmp < 0;
    //             break;
    //         case OP_GT:
    //             res = cmp > 0;
    //             break;
    //         case OP_LE:
    //             res = cmp <= 0;
    //             break;
    //         case OP_GE:
    //             res = cmp >= 0;
    //             break;
    //         }

    //         if (!res)
    //         {
    //             return false;
    //         }

    //     }
    //     return true;
    // }
    // //初始化缓冲区
    // void init_buffer()
    // {
    //     left_buffer.clear();
    //     left_->beginTuple();
    //     while (!left_->is_end() && left_buffer.size() < LEFT_BUFFER_SIZE)
    //     {
    //         left_buffer.emplace_back(left_->Next());
    //         left_->nextTuple();
    //     }
    //     left_buffer_index = 0;
    //     if (left_buffer.size() == 0 && left_->is_end())
    //     {
    //         isend = true;
    //         return;
    //     }

    //     right_buffer.clear();
    //     right_->beginTuple();
    //     while (!right_->is_end() && right_buffer.size() < RIGHT_BUFFER_SIZE)
    //     {
    //         right_buffer.emplace_back(right_->Next());
    //         right_->nextTuple();
    //     }
    //     right_buffer_index = 0;
    //     if (right_buffer.size() == 0 && right_->is_end())
    //     {
    //         isend = true;
    //         return;
    //     }
    // }

    //试试优化
    //主要思路 用两个buffer表示两个表 随后buffer大小就对应一个块，随后按照嵌套块循环连接的方法实现
    void beginTuple() override {
        //初始化两个缓存
        init_buffer();
        if (isend)
            return;
        while (true)
        {
            //外部块元组遍历
            for (; left_buffer_index < left_buffer.size(); left_buffer_index++)
            {
                //内部块元组遍历
                for (; right_buffer_index < right_buffer.size(); right_buffer_index++)
                {
                    if (test_join_conds(*left_buffer[left_buffer_index], *right_buffer[right_buffer_index]))
                    {
                        //找到了符合条件的第一个
                        return;
                    }
                }
                //当前内部块缓存被全部扫描，置为0
                right_buffer_index = 0;
            }

            left_buffer_index = 0;

            //内部块缓存满了，读入下一个块
            right_buffer.clear();

            while (!right_->is_end() && right_buffer.size() < RIGHT_BUFFER_SIZE)
            {
                right_buffer.emplace_back(right_->Next());
                right_->nextTuple();
            }

            //内部块遍历完毕，读入外部快的下一块
            if (right_buffer.size() == 0)
            {
                left_buffer.clear();
                while (!left_->is_end() && left_buffer.size() < LEFT_BUFFER_SIZE)
                {
                    left_buffer.emplace_back(left_->Next());
                    left_->nextTuple();
                }
                if (left_buffer.size() == 0 && left_->is_end())
                {
                    isend = true;
                    return;
                }
                left_buffer_index = 0;

                //重新读入内部块
                right_->beginTuple();
                while (!right_->is_end() && right_buffer.size() < RIGHT_BUFFER_SIZE)
                {
                    right_buffer.emplace_back(right_->Next());
                    right_->nextTuple();
                }

            }
        }
        // while(true)
        // {
        //     //外部块元组遍历
        //     while(left_buffer_index < left_buffer.size())
        //     {
        //         //内部块元组遍历
        //         while(right_buffer_index < right_buffer.size())
        //         {
        //             //满足条件则直接返回 因已排序，不满足则内部指针++
        //             if(test_join_conds(*left_buffer[left_buffer_index], *right_buffer[right_buffer_index]))
        //                 return;
        //             else right_buffer_index++;
        //         }
        //         //内部块遍历完毕 重置

        //     }
        // }
    }

    void nextTuple() override
    {
        right_buffer_index++;
        while (true)
        {
            //外部块元组遍历
            for (; left_buffer_index < left_buffer.size(); left_buffer_index++)
            {
                //内部块元组遍历
                for (; right_buffer_index < right_buffer.size(); right_buffer_index++)
                {
                    if (test_join_conds(*left_buffer[left_buffer_index], *right_buffer[right_buffer_index]))
                    {
                        //找到了符合条件的第一个
                        return;
                    }
                }
                //当前内部块缓存被全部扫描，置为0
                right_buffer_index = 0;
            }

            left_buffer_index = 0;

            //内部块缓存满了，读入下一个块
            right_buffer.clear();

            while (!right_->is_end() && right_buffer.size() < RIGHT_BUFFER_SIZE)
            {
                right_buffer.emplace_back(right_->Next());
                right_->nextTuple();
            }

            //内部块遍历完毕，读入外部快的下一块
            if (right_buffer.size() == 0)
            {
                left_buffer.clear();
                while (!left_->is_end() && left_buffer.size() < LEFT_BUFFER_SIZE)
                {
                    left_buffer.emplace_back(left_->Next());
                    left_->nextTuple();
                }
                if (left_buffer.size() == 0 && left_->is_end())
                {
                    isend = true;
                    return;
                }
                left_buffer_index = 0;

                //重新读入内部块
                right_->beginTuple();
                while (!right_->is_end() && right_buffer.size() < RIGHT_BUFFER_SIZE)
                {
                    right_buffer.emplace_back(right_->Next());
                    right_->nextTuple();
                }

            }
        }
    }
    //用于检验连接条件
    bool test_join_conds(RmRecord& lrecord, RmRecord& rrecord)
    {
        // for (auto& cond : fed_conds_)
        int cond_num=fed_conds_.size();
        for (int i=0;i<cond_num;++i)
        {
            char* lval = lrecord.data + cond_offset[i].first;
            char* rval = rrecord.data + cond_offset[i].second;

            int cmp;
            //比较方法与抽象类中定义的一致
            switch (cond_lhs_type[i])
            {
            case TYPE_INT:
            {
                int il = *(int*)lval;
                int ir = *(int*)rval;
                // cmp=il-ir;
                cmp = (il < ir) ? -1 : ((il > ir) ? 1 : 0);
                break;
            }
            case TYPE_BIGINT:
            {
                long long bl = *(long long*)lval;
                long long br = *(long long*)rval;
                cmp = (bl < br) ? -1 : ((bl > br) ? 1 : 0);
                break;
            }
            case TYPE_FLOAT:
            {
                float fl = *(float*)lval;
                float fr = *(float*)rval;
                cmp = (fl < fr) ? -1 : ((fl > fr) ? 1 : 0);
                break;
            }
            case TYPE_STRING:
            {
                std::string sl = *(std::string*)lval;
                std::string sr = *(std::string*)rval;
                cmp = sl.compare(sr);
                break;
            }
            case TYPE_TIME:
            {
                cmp = memcmp(lval, rval, 19);
            }
            }

            bool res = true;
            switch (fed_conds_[i].op)
            {
            case OP_EQ:
                res = cmp == 0;
                break;
            case OP_NE:
                res = cmp != 0;
                break;
            case OP_LT:
                res = cmp < 0;
                break;
            case OP_GT:
                res = cmp > 0;
                break;
            case OP_LE:
                res = cmp <= 0;
                break;
            case OP_GE:
                res = cmp >= 0;
                break;
            }

            if (!res)
            {
                return false;
            }

        }
        //没有条件就是全连接的笛卡尔积 
        return true;
    }
    //初始化缓冲区
    void init_buffer()
    {
        left_buffer.clear();
        left_->beginTuple();
        while (!left_->is_end() && left_buffer.size() < LEFT_BUFFER_SIZE)
        {
            left_buffer.emplace_back(left_->Next());
            left_->nextTuple();
        }
        left_buffer_index = 0;
        if (left_buffer.size() == 0 && left_->is_end())
        {
            isend = true;
            return;
        }

        right_buffer.clear();
        right_->beginTuple();
        while (!right_->is_end() && right_buffer.size() < RIGHT_BUFFER_SIZE)
        {
            right_buffer.emplace_back(right_->Next());
            right_->nextTuple();
        }
        right_buffer_index = 0;
        if (right_buffer.size() == 0 && right_->is_end())
        {
            isend = true;
            return;
        }
        // //排序以便优化
        // sort_buffer(left_buffer);
        // sort_buffer(right_buffer);
        
    }
    // //根据条件列将缓冲区排序
    // void sort_buffer(std::vector<std::unique_ptr<RmRecord>> &buffer)
    // {
    //     std::sort(buffer.begin(), buffer.end(), [this](const std::unique_ptr<RmRecord>& r1, const std::unique_ptr<RmRecord>& r2) {
    //     // for(auto& single_col:cols_)
    //         int cond_num=fed_conds_.size();
    //         for (int i=0;i<cond_num;++i)
    //         {
    //             char* lval = lrecord.data + cond_offset[i].first;
    //             char* rval = rrecord.data + cond_offset[i].second;
    
    //             int cmp=-1;
    //             //比较方法与抽象类中定义的一致
    //             switch (cond_lhs_type[i])
    //             {
    //                 case TYPE_INT:
    //                 {
    //                     int il = *(int*)lval;
    //                     int ir = *(int*)rval;
    //                     // cmp=il-ir;
    //                     cmp = (il < ir) ? -1 : ((il > ir) ? 1 : 0);
    //                     break;
    //                 }
    //                 case TYPE_BIGINT:
    //                 {
    //                     long long bl = *(long long*)lval;
    //                     long long br = *(long long*)rval;
    //                     cmp = (bl < br) ? -1 : ((bl > br) ? 1 : 0);
    //                     break;
    //                 }
    //                 case TYPE_FLOAT:
    //                 {
    //                     float fl = *(float*)lval;
    //                     float fr = *(float*)rval;
    //                     cmp = (fl < fr) ? -1 : ((fl > fr) ? 1 : 0);
    //                     break;
    //                 }
    //                 case TYPE_STRING:
    //                 {
    //                     std::string sl = *(std::string*)lval;
    //                     std::string sr = *(std::string*)rval;
    //                     cmp = sl.compare(sr);
    //                     break;
    //                 }
    //                 case TYPE_TIME:
    //                 {
    //                     cmp = memcmp(lval, rval, 19);
    //                 }
    //             }
    //             return cmp<=0;
    //         }
    //     });
    // }

    std::unique_ptr<RmRecord> Next() override {
        auto record = std::make_unique<RmRecord>(len_);
        memcpy(record->data, left_buffer[left_buffer_index]->data, left_->tupleLen());
        memcpy(record->data + left_->tupleLen(), right_buffer[right_buffer_index]->data, right_->tupleLen());
        return record;
    }

    size_t tupleLen() const override { return len_; }

    const std::vector<ColMeta>& cols() const override { return cols_; }

    bool is_end() const override { return isend; }

    Rid& rid() override { return _abstract_rid; }
};